#include "Connection.h"
#include <assert.h>
#include "EventLoop.h"
#include "Channel.h"
#include "Socket.h"
#include "Log.h"

namespace dsth {

Connection::Connection(EventLoop* loop, int id, int socket_fd,
        const InetSocketAddress& host_addr, const InetSocketAddress& peer_addr) 
    : loop_(loop),
    socket_(new Socket(socket_fd)),
    channel_(new Channel(socket_->GetFd(), loop)),
    id_(id),
    host_addr_(host_addr), 
    peer_addr_(peer_addr) {
    channel_->SetReadHandler(std::make_pair(this, &Connection::OnReadEnable));
    channel_->SetWriteHandler(std::make_pair(this, &Connection::OnWriteEnable));
    channel_->SetErrorHandler(std::make_pair(this, &Connection::OnErrorOccur));
}


Connection::~Connection() {
    LOG_DEBUG("[Connection::~Connection] id:%d", id_);
    loop_->Remove(channel_.get());
}

void Connection::ConnectEstablished() {
    //assert(loop_->IsInLoopThread());
    channel_->EnableReading();
    conn_handler_(this);
}

void Connection::ConnectDestroy() {
    assert(loop_->IsInLoopThread());
    loop_->AddErasedConnection(this);
}

void Connection::ShutDown() {
    LOG_DEBUG("[Connection::ShutDown]");
    socket_->ShutDownWrite();
}

void Connection::OnReadEnable(Timestamp recv_time) {
    int socket_err = 0;
    int n = inbuffer_.ReadFd(socket_->GetFd(), &socket_err);
    LOG_DEBUG("[Connection::OnReadEnable] ReadFd return:%d.", n);
    if (n == 0) {
        OnClose();
    } 
    else if (n > 0) {
        message_handler_(this, &inbuffer_, recv_time);
    }
    else {
        if (socket_err != WSAECONNRESET)
            LOG_ERROR("[Connection::OnReadEnable] readv error:%d fd:%d", socket_err, socket_->GetFd());
        OnClose();
    }
}

void Connection::OnWriteEnable() {
    LOG_TRACE("[Connection::OnWriteEnable]");

    if (!channel_->IsWriting()) {
        LOG_TRACE("clhannel is closed.");
        return;
    }

    int wrote_bytes = outbuffer_.WriteFd(socket_->GetFd());
    if (wrote_bytes > 0) {
        if (wrote_bytes == static_cast<int>(outbuffer_.ReadableBytes())) {
            channel_->DisableWriting();
            write_completed_handler_(this);
        }
        else {
            LOG_DEBUG("[Connection::OnWriteEnable] wrote byte:%d buf readable:%d, will write more.\n",
                wrote_bytes, outbuffer_.ReadableBytes());
        }
        outbuffer_.Retrieve(wrote_bytes);
    }
    else {
        HandleSocketError();
    }
}

void Connection::OnClose() {
    assert(loop_->IsInLoopThread());
    LOG_TRACE("[Connection::OnClose]");
    channel_->DisableAllEvent();
    // don't remove channel here.
    dis_conn_handler_(this);
}

void Connection::OnErrorOccur() {
    LOG_ERROR("[Connection::OnErrorOccur]");
}

//void Connection::Send(const char* data, size_t len) {
//    if (loop_->IsInLoopThread()) {
//        SendInLoop(data, len);
//    }
//    else {
//        loop_->RunInLoop(std::make_pair(this, &Connection::SendInLoop));
//    }
//}

//TODO: when run this function in work thread loop, call this in its socket thread loop
void Connection::Send(const char* data, size_t len) {
    //assert(loop_->IsInLoopThread());
    int wrote_byte = 0;
    if (!channel_->IsWriting() && outbuffer_.ReadableBytes() == 0) {
        LOG_TRACE("[Connection::Send] channel.IsWriting:%d, outbuffer.ReadableBytes:%d.",
            channel_->IsWriting(), outbuffer_.ReadableBytes());
        wrote_byte = net::Write(channel_->GetSocketFd(), data, len);
        if (wrote_byte >= 0) {
            if (wrote_byte < static_cast<int>(len)) {
                LOG_DEBUG("[Connection::Send] wrote byte:%d len:%d, will write more.\n", wrote_byte, len);
            }
            else {
                LOG_DEBUG("[Connection::Send] wrote byte:%d len:%d, writing complete.\n", wrote_byte, len);
                write_completed_handler_(this);
            }
        }
        else {
            HandleSocketError();
            wrote_byte = 0;
        }
    }
    if (wrote_byte  < static_cast<int>(len)) {
        outbuffer_.Append(data + wrote_byte, len - wrote_byte);
        channel_->EnableWriting();
    }
}

void Connection::Send(const ChannelBuffer& buf) {
    Send(buf.Peek(), buf.ReadableBytes());
}

void Connection::Send(const std::string& data) {
    Send(data.c_str(), data.size());
}

void Connection::HandleSocketError() {
#ifdef IS_WINDOWS
    int err = ::WSAGetLastError();
    if (err == WSAEWOULDBLOCK) {
        return;
    }
    if (err == WSAECONNRESET) {
        LOG_DEBUG("[Connection::HandleSocketError] WSAECONNRESET errorcode:%d id:%d\n", err, id_);
    }
    else {
        LOG_ERROR("[Connection::HandleSocketError] write socket failed, errorcode:%d\n", err);
    }
#else// TODO: for POSIX
    int err = errno;
    if (err != EWOULDBLOCK) {
        LOG_ERROR("[Connection::Send] write socket failed, errorcode:%d\n", err);
    }
#endif
    channel_->DisableAllEvent();
    dis_conn_handler_(this);
}

} // namespace
